package com.dd.ms.mw.kafka.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author liuxianmeng
 * @CreateTime 2025/6/9 0:48
 * @Description kafkaTemplateConfig
 */
@Configuration
public class KafkaProducerConfig {

    @Autowired
    private KafkaConfigProperties kafkaConfigProperties;

    // 如果使用Avro可取消注释
    //@Value("${schema.registry.url}")
    //private String schemaRegistryUrl;

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfigProperties.getBootstrapServers());
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        // Avro序列化配置（按需启用）
        // configProps.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

        /**
         🟥[TRUSTED_PACKAGES必须指定 否则报错] 添加JsonDeserializer的信任包配置 JsonUserBO所在的包
         🟢[TRUSTED_PACKAGES如何指定多个包？]
         (1) 逗号间隔
         props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.dd.stack.kafka.common.pojo.dto,com.example.pkg1,com.example.pkg2");
         (2) 使用通配符
         props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.dd.stack.kafka.*,org.example.*");
         */
        configProps.put(JsonDeserializer.TRUSTED_PACKAGES, kafkaConfigProperties.getTrustedPackages());

        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplateObj() {
        return new KafkaTemplate<>(producerFactory());
    }
}
